package com.hmc.kafka.consumer.group;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.RoundRobinAssignor;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * 分区策略：默认使用range+roundRobin的策略
 * range策略：在一个topic中，所有分区数除消费者数，如果除不尽，则前面的消费着多承担几个分区
 *   例子：topic中分区为 0，1，2，3，4，5，6 消费者组中消费者为 a,b,c
 *      分配策略 ： 7/3=2.。。。1 所以 a消费0，1，2 b消费 4，3 c消费6，5
 *      在45s内，当一个消费者挂掉之后，由他所消费的分区将整体交给一个消费者
 *      45s以后，将重新分配分区
 * roundRobin策略：将消费的所有topic的所有分区数进行轮询分配
 * 需要配置partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor
 *   例子：topic1中分区为t1p0,t1p1,t1p2,t1p3 topic2中分区为t2p0,t2p1,t2p2 消费者 a,b,c
 *      分配策略：a:t1p0,t1p3,t2p2  b:t1p1,t2p0  c:t1p2,t2p1
 *      当一个消费者挂掉后，它所消费的分区使用轮询策略分发给其他节点
 * stick策略：将分区均匀的分发给消费者，每个消费的所承担的分区是随机的
 */
public class CustomConsumer {
    public static void main(String[] args) {
        //创建配置类
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        //配置组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"0");
        //配置分区分配策略
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());
        //创建消费者
        KafkaConsumer<String,String> consumer = new KafkaConsumer(properties);
        //订阅topic
        List<String> list = Arrays.asList("second","third");
        consumer.subscribe(list);
        //消费数据
        while (true) {
            //拉取数据，每一秒钟一次
            ConsumerRecords<String, String> message = consumer.poll(Duration.ofSeconds(1));
            message.iterator().forEachRemaining(a->{
                System.out.println(a);
            });
        }
    }
}
